package com.rizzo.shell.plugins;

import com.google.common.collect.ImmutableMap;
import com.rabbitmq.client.Channel;
import com.rizzo.shell.intern.AmqpCommands;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.logging.Logger;

@Configuration
public class AmqpConfig {

	private static Logger logger = Logger.getLogger(AmqpConfig.class.getCanonicalName());

	@Resource(name = "args")
	private HashMap<AmqpCommands.TransferArgument, String> args;

	@Bean(name = "rabbitConnectionFactory")
	public CachingConnectionFactory rabbitConnectionFactory() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
				args.get(AmqpCommands.TransferArgument.HOST), Integer.valueOf(args.get(AmqpCommands.TransferArgument.PORT)));
		connectionFactory.setPublisherConfirms(true);
		connectionFactory.setUsername(args.get(AmqpCommands.TransferArgument.USERNAME));
		connectionFactory.setPassword(args.get(AmqpCommands.TransferArgument.PASSWORD));
		connectionFactory.setVirtualHost(args.get(AmqpCommands.TransferArgument.VHOST));
		return connectionFactory;
	}

	@Bean(name = "amqpAdmin")
	public AmqpAdmin amqpAdmin() {
		return new RabbitAdmin(rabbitConnectionFactory());
	}

	@Bean(name = "rabbitTemplate")
	public RabbitTemplate rabbitTemplate() {
		final RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory());
		return rabbitTemplate;
	}

	@Bean(name = "myQueue")
	public Queue myQueue() {
		final ImmutableMap<String, Object> immutableMap = ImmutableMap.<String, Object>of(
//				"x-message-ttl", Long.parseLong(args.get(AmqpCommands.TransferArgument.TTL)),
				"x-ha-policy", "all");
		return new Queue("my-queue", true, false, true, immutableMap);
	}

	@Bean(name = "messageListener")
	public ChannelAwareMessageListener channelAwareMessageListener() {
		ChannelAwareMessageListener channelAwareMessageListener = new ChannelAwareMessageListener() {
			@Override
			public void onMessage(Message message, Channel channel) throws Exception {
//				logger.log(Level.INFO, "Channel :" + channel.toString() + " - Consumed message: " + message.toString());
//				if (Integer.parseInt(new String(message.getBody())) % 100 != 0) {
				channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
//				} else {
//					logger.log(Level.INFO, "Not Acking message: " + message.getMessageProperties().getDeliveryTag());
//				}
			}
		};
		return channelAwareMessageListener;
	}

	@Bean(name = "topicExchange")
	public TopicExchange topicExchange() {
		return new TopicExchange("logging");
	}

	@Bean(name = "binding")
	public Binding binding() {
		return BindingBuilder.bind(myQueue()).to(topicExchange()).with("my-queue-key");
	}

//	@Bean(name = "rabbitTransactionManager")
//	public RabbitTransactionManager rabbitTransactionManager() {
//		final RabbitTransactionManager rabbitTransactionManager = new RabbitTransactionManager(rabbitConnectionFactory());
//		return rabbitTransactionManager;
//	}

	@Bean(name = "messageListenerContainer")
	public SimpleMessageListenerContainer getMessageListenerContainer() {
		final SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(rabbitConnectionFactory());
		simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//		simpleMessageListenerContainer.setChannelTransacted(true);
		simpleMessageListenerContainer.setConcurrentConsumers(10);
		simpleMessageListenerContainer.setQueues(myQueue());
		simpleMessageListenerContainer.setMessageListener(channelAwareMessageListener());
//		simpleMessageListenerContainer.setTransactionManager(rabbitTransactionManager());
		return simpleMessageListenerContainer;
	}

}
